package com.atguigu.tingshu.album.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.AlbumStatVoMsg;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;

@CanalEventListener
@Slf4j
public class CanalClientListener {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private KafkaService kafkaService;

    //监听album_info表
    @ListenPoint(destination = "example", schema = "tingshu_album", table = {"album_info"})
    public void albumInfoUpdateListener(CanalEntry.EntryType entryType, CanalEntry.RowData rowChange) {
        List<CanalEntry.Column> beforeColumnsList = rowChange.getBeforeColumnsList();
        beforeColumnsList.forEach(column -> {
            String name = column.getName();
            if ("id".equals(name)) {
                String value = column.getValue();
                stringRedisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX + value);
                log.info("从redis中删除了专辑的基本数据,albumid:{}", value);
            }
        });
    }

    //监听album_stat表
    @ListenPoint(destination = "example", schema = "tingshu_album", table = {"album_stat"})
    public void albumStatUpdateListener(CanalEntry.EntryType entryType, CanalEntry.RowData rowChange) {
        AlbumStatVoMsg albumStatVoMsg = new AlbumStatVoMsg();
        List<CanalEntry.Column> beforeColumnsList = rowChange.getBeforeColumnsList();
        for (CanalEntry.Column column : beforeColumnsList) {
            String name = column.getName();
            String value = column.getValue();
            if ("album_id".equals(name)) {
                stringRedisTemplate.delete(RedisConstant.ALBUM_STAT_PREFIX + value);
                albumStatVoMsg.setAlbumId(Long.getLong(value));
                log.info("从redis中删除了专辑的基本数据,albumid:{}", value);
            }
            if ("stat_num".equals(name)) {
                albumStatVoMsg.setStatNum(Integer.getInteger(value));
            }
            if ("stat_type".equals(name)) {
                albumStatVoMsg.setStatType(value);
            }
        }
        //同步ES中的数据
        //使用kafka发送消息
        //确认发送消息内容 id statType statNum;
        kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_STAT_UPDATE_SYNC, JSON.toJSONString(albumStatVoMsg));
        log.info("kafka发送同步数据到ES的消息...{}",JSON.toJSONString(albumStatVoMsg) );
    }

    @ListenPoint(destination = "example", schema = "tingshu_album", table = {"base_category3"})
    public void baseCategoryViewUpdateListener(CanalEntry.EntryType entryType, CanalEntry.RowData rowChange) {
        List<CanalEntry.Column> afterColumnsList = rowChange.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            String columnValue = column.getValue();
            if("id".equals(columnName)){
                String redisKey = RedisConstant.CATEGORY_INFO_SUFFIX + columnValue;
                stringRedisTemplate.delete(redisKey);
                log.info("分类数据已删除 redisKey: {}",redisKey);
            }
        });

    }
    @ListenPoint(destination = "example", schema = "tingshu_album", table = {"user_info"})
    public void UserInfoUpdateListener(CanalEntry.EntryType entryType, CanalEntry.RowData rowChange) {
        List<CanalEntry.Column> afterColumnsList = rowChange.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            String columnValue = column.getValue();
            if("id".equals(columnName)){
                stringRedisTemplate.delete(RedisConstant.USER_INFO_PREFIX+ columnValue);
                log.info("分类数据已删除 userId: {}",columnValue);
            }
        });

    }
}
